package dataflowscheme.engine.core;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.log4j.Logger;

/**
 * Execution Event Parser thread is processing the Execution Event queue - 
 * - in each iteration it takes one event (or thread waits until an event
 * is available) and invoke its processing dependent on a type of an event. 
 * The processing of each event is performed in a separate thread.
 * 
 * @author <a href="mailto:misiak7@gmail.com">Michal Antolik</a>
 *
 */
public class ExecutionEventParser implements Runnable {

	static Logger log = Logger.getLogger(ExecutionEventParser.class);
	
	/** queue where are events of type <code>IExecutionEvent<code> put  */
	private final LinkedBlockingQueue<IExecutionEvent> queue;
	
	/** instance of the Activity checker to check state of processed events*/
	private final ActivityChecker activityChecker;
	
	/** when true, then parser will end processing */
	private AtomicBoolean stop = new AtomicBoolean(false);
	
	/** thread pool for processing events in separate thread */
	private final ExecutorService executor = Executors.newCachedThreadPool();
	
	/** if statistics should be recorded */
	private final boolean printStats;
	
	public ExecutionEventParser(LinkedBlockingQueue<IExecutionEvent> executionQueue, ActivityChecker activityChecker, boolean prinStats) {
		this.queue = executionQueue;
		this.activityChecker = activityChecker;
		this.printStats = prinStats;
	}
	
	public void run() {
		try{
			while(!stop.get()){
				IExecutionEvent e = queue.take();
				Future<?> pe = e.execute(this, executor, printStats);
				activityChecker.add(pe); //add event to observe its state
			}
			if(queue.size() != 0){
				log.warn("Ending - remaining unprocessed events: " + queue.size());
			}
			executor.shutdown();
			
			long BREAKTIME = 1000;
			int numberOfbreaks = 0;
			while(!executor.isTerminated() && (numberOfbreaks < 10)){
				//wait until all submit events are processed
				log.info("Waiting for shutdown of execution parser... attempt " + numberOfbreaks++);
				Thread.sleep(BREAKTIME);
			}
		}catch(InterruptedException ie){
			log.debug("Executor interrupted", ie);
		} finally {
			if(!executor.isTerminated()){
				log.debug("ThreadExecutor is going to be shutdown NOW (last attempt).");
				executor.shutdownNow();
			}
			log.debug("I'm leaving, arrivederci!");
		}

	}
	
	/**
	 * stop the parser - it is called from the execution end event
	 */
	public void stop(){
		stop.set(true);
	}


}
